package edu.gatech.ors.mrsim.core.impl;

import java.util.HashMap;

import edu.gatech.ors.mrsim.core.Cluster;
import edu.gatech.ors.mrsim.core.ModuleOnNode;
import edu.gatech.ors.mrsim.core.Node;
import edu.gatech.ors.mrsim.core.Simulation;
import edu.gatech.ors.mrsim.io.conf.Configuration;

public class HDFS extends ModuleOnNode<Node> {
	public static final String NAME = "HDFS";
	private int numReplication;
	private double bandwidth;

	private HashMap<String, HDFSNameNode> nameNodes;
	private HashMap<String, HDFSDataNode> dataNodes;

	public HDFS(Simulation simulation, Configuration config) {
		super(simulation, config);
		nameNodes = new HashMap<String, HDFSNameNode>();
		dataNodes = new HashMap<String, HDFSDataNode>();
		Cluster cluster = simulation.getCluster();
		for (Node node : cluster.getNodes()) {
			if (node instanceof HDFSNameNode)
				nameNodes.put(node.getName(), (HDFSNameNode) node);
			if (node instanceof HDFSDataNode)
				dataNodes.put(node.getName(), (HDFSDataNode) node);
		}

	}

	@Override
	public boolean loadValuesFromConfiguration() {
		configuration = checkForRedirect(LOGGER);
		boolean valid = true;
		try {
			numReplication = configuration.getInt("numReplication");
			bandwidth = configuration.getDouble("bandwidth");
			String nodeName = configuration.getString("HDFS host");
			host = simulation.getCluster().getNode(nodeName);
		} catch (NullPointerException e) {
			valid = false;
			LOGGER.error("Could not load a config value in HDFS. Required parameter not provided as key.", e);
		}
		return valid;
	}

	@Override
	public String getName() {
		return NAME;
	}

	public boolean handleIORequest(IORequest event) {
		String node = event.getNodeName();
		if (dataNodes.containsKey(node))
			if (dataNodes.get(node).respondToRequest(event)) {
				event.addDuration((long) (bandwidth * event.getData().getSize()));
				event.setCompleted(true);
				this.propagateEvent(event);
				return true;
			}
		event.setCompleted(false);
		this.propagateEvent(event);
		return true;
	}

	public boolean handleFileRequest(FileRequest event) {
		if (event.isWrite()) {
			String[] nodes = new String[numReplication];
			// List<Integer> selected=new ArrayList<Integer>();
			for (int i = 0; i < numReplication; i++) {
				// nodes[i]=
			}
		}
		for (HDFSNameNode nameNode : nameNodes.values()) {
			if (nameNode.hasFile(event.getFileName())) {
				event.setResponse(nameNode.getFileLocation(event.getFileName()));
				event.setCompleted(true);
				event.setDuration((int) (Math.random() * 10));
				propagateEvent(event);
				return true;
			}

			return true;
		}
		event.setCompleted(false);
		event.setDuration((int) (Math.random() * 10));
		propagateEvent(event);
		return true;
	}
}
